package com.business.etl.flow.process;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.business.etl.flow.status.StatusManager;
import com.google.common.graph.Graph;
import com.component.api.AbstractComponent;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.constants.FlowStatus;
import com.component.api.constants.NodeStatus;
import com.component.api.excepiton.FlowRunningException;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.DatabaseParamDTO;
import com.component.api.model.flow.NodeInfo;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.ParamType;
import com.component.api.model.param.TableColumn;
import com.component.api.utils.ComponentUtils;
import com.common.utils.constant.SQLConstants;
import com.etl.component.common.AbstractFlinkComponent;
import com.service.database.action.JDBCExecutor;
import com.service.database.config.DataBase;
import com.service.database.config.DatabaseConfig;
import com.service.database.model.DataColumn;
import com.common.log.Logger;
import lombok.Data;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.util.*;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * 描述：
 * 执行单节点线程
 * @author xianggj
 * @Date 2021/10/15 16:43
 **/
@Data
public class ExecNode implements Supplier<ProcessResult> {

    /**
     * 当前节点信息
     */
    private NodeInfo nodeInfo;

    /**
     * 当前节点信息
     */
    private FlowContext flowContext;

    /**
     * 当前节点信息
     */
    private Logger logger;

    public ExecNode(NodeInfo nodeInfo,  FlowContext flowContext, Logger logger) {
        this.nodeInfo = nodeInfo;
        this.flowContext = flowContext;
        this.logger = logger;
    }

    @Override
    public ProcessResult get() {
        logger.info("执行前准备...");
        ProcessHandler processHandler = flowContext.getProcessHandler();
        Graph<NodeInfo> graph = flowContext.getGraph();
        StatusManager statusManager = flowContext.getStatusManager();
        //记录线程
        processHandler.putThread(nodeInfo.getId(), Thread.currentThread());
        Set<NodeInfo> predecessors = graph.predecessors(nodeInfo);
        List<PortData> preDatas = new ArrayList<>();
        //不为空 先判断前提条件
        if (!predecessors.isEmpty()){
            for (NodeInfo preNode : predecessors) {
                if (processHandler.isCompleted(preNode.getId())){
                    ProcessResult processResult = processHandler.getResult(preNode.getId());
                    preDatas.addAll(processResult.getPortDatas());
                } else {
                    //如果先天条件不满足 直接挂起线程 并记录线程
                    String log = "因为依赖节点【"+preNode.getName()+"】未运行成功，" +
                            "节点【" +nodeInfo.getName()+"】进入等待...";
                    logger.info(log);
                    //方便打印日志
                    Thread.currentThread().setName(nodeInfo.getName());
                    //等待状态
                    statusManager.setNodeStatus(flowContext.getFlowInfo().getFlowId(),
                            nodeInfo.getId(), NodeStatus.WAIT);
                    //挂起线程 耗时操作 有可能这段时间那个节点已经运行完了
                    //记录线程依赖
                    if (processHandler.putRelyNoThread(preNode.getId(), Thread.currentThread())){
                        //阻塞当前线程，直到deadline【从1970年开始到deadline时间的毫秒数】时间
                        LockSupport.parkUntil(System.currentTimeMillis() + processHandler.MAX_TIME);
                    }

                    //恢复后继续添加节点
                    if (processHandler.isCompleted(preNode.getId())){
                        ProcessResult processResult = processHandler.getResult(preNode.getId());
                        preDatas.addAll(processResult.getPortDatas());
                    }
                }
            }
        }
        logger.info("参数校验...");
        String flowStatus = statusManager.getFlowStatus(flowContext.getFlowInfo().getFlowId());
        //流程已经停止
        if (FlowStatus.STOP.name().equalsIgnoreCase(flowStatus)){
            statusManager.setNodeStatus(flowContext.getFlowInfo().getFlowId(),
                    nodeInfo.getId(), NodeStatus.STOP);
            return new ProcessResult(flowContext.getFlowInfo().getFlowId(), nodeInfo.getId());
        }
        //等待状态
        statusManager.setNodeStatus(flowContext.getFlowInfo().getFlowId(),
                nodeInfo.getId(), NodeStatus.RUNNING);
        TimeInterval timer = DateUtil.timer();
        String rule = nodeInfo.getRule();
        String param = nodeInfo.getParam();
        List<ParamPair> paramPairs = null;
        try {
            //参数格式不做限制
            paramPairs = JSON.parseArray(param, ParamPair.class);
        } catch (Exception e) {
            e.printStackTrace();
            throw new FlowRunningException("参数解析错误"+e.getLocalizedMessage());
        }
        logger.info("参数解析完成.");
        AbstractComponent component = ComponentUtils.getComponentByRule(rule);
        //对象唯一
        component.setFlowKey(flowContext.hashCodeStr());
        component.setNodeId(nodeInfo.getId());
        //尾节点标记执行
        component.setNeedExec(nodeInfo.getIsExec());

        //数据输出提前创表
        if (StrUtil.endWith(rule, ComponentCodeConstants._OUTPUT)){
            logger.info("检查表信息是否完善...");
            List<String> ids = null;
            try {
                ids = checkTableAndPerfect(preDatas, paramPairs, component);
            } catch (Exception e) {
                e.printStackTrace();
                throw new FlowRunningException("创表异常"+e.getMessage());
            }
            if (ids != null){
                paramPairs.add(new ParamPair(ComponentConstants.IDS, ParamType.ARRAY,
                        JSON.toJSONString(ids), String.class));
            }
            logger.info("表信息验证完成.");
        }
        logger.info("开始执行...");
        ProcessResult process = component.process(preDatas, paramPairs);
        process.setNodeId(nodeInfo.getId());
        logger.info("执行结束.");
        logger.info("执行耗时" + timer.interval() + "ms");
        //等待状态
        statusManager.setNodeStatus(flowContext.getFlowInfo().getFlowId(),
                nodeInfo.getId(), NodeStatus.FINISH);
        return process;
    }

    /**
     * 检查数据输出是否完善
     * @param preDatas
     * @param paramPairs
     * @param component
     */
    private List<String> checkTableAndPerfect(List<PortData> preDatas, List<ParamPair> paramPairs, AbstractComponent component) {
        Map<String, Object> params = component.paramToObject(paramPairs);
        DatabaseParamDTO databaseParamDTO = (DatabaseParamDTO)params.get(ComponentConstants.CONFIG);

        String writeModel = (String)params.get(ComponentConstants.WRITE_MODEL);
        //覆盖策略 先删除表
        DatabaseConfig databaseConfig = getDatabaseConfig(databaseParamDTO);
        if (ComponentConstants.OVERWRITE.equalsIgnoreCase(writeModel)){
            try {
                JDBCExecutor.dropTable(databaseConfig);
            } catch (Exception e) {
                e.printStackTrace();
                throw new FlowRunningException("覆盖表失败:"+e.getMessage());
            }
        }

        SetData input = (SetData)preDatas.get(0).getValue();
        DataSet<Row> result = (DataSet<Row>) input.getDataset();
        List<TableColumn> columns = input.getColumns();
        RowTypeInfo type = (RowTypeInfo)result.getType();
        TypeInformation[] fieldTypeInfos = type.getFieldTypes();
        if (fieldTypeInfos.length != columns.size()){
            throw new FlowRunningException("数据长度和字段长度不一致");
        }
        //主键
        String id = SQLConstants.ID;
        String idStr = null;
        List<DataColumn> cols = new ArrayList<>();
        for (int i = 0; i < columns.size(); i++) {
            TableColumn column = columns.get(i);
            DataColumn dataColumn = new DataColumn();
            String columnName = column.getColumnName();
            dataColumn.setDataType(column.getDataType());
            dataColumn.setColumnName(columnName);
            dataColumn.setColumnComment(column.getColumnComment());
            dataColumn.setColumnKey(column.getColumnKey());
            //避免id大小写问题
            if (SQLConstants.ID.equalsIgnoreCase(columnName)){
                id = columnName;
            }
            if (dataColumn.getColumnKey()){
                idStr = columnName;
            }
            dataColumn.setDataTypeInt(((AbstractFlinkComponent)component).typeToInt(fieldTypeInfos[i]));
            dataColumn.setIsNull(column.getIsNull());
            dataColumn.setLength(column.getLength());
            dataColumn.setFloatLength(column.getFloatLength());
            cols.add(dataColumn);
        }
        if (idStr == null ){
            idStr = id;
        }
        try {
            JDBCExecutor.createTable(databaseConfig, cols);
        } catch (Exception e) {
            e.printStackTrace();
            throw new FlowRunningException("创建表失败:"+ e.getMessage());
        }
        if (ComponentConstants.INSERT_UPDATE.equalsIgnoreCase(writeModel)){
            try {
                List<Object> objects = JDBCExecutor.distinctCol(databaseConfig, idStr);
                List<String> value = objects.stream()
                        .filter(Objects::nonNull)
                        .map(e -> e.toString()).collect(Collectors.toList());
                return value;
            } catch (Exception e) {
                e.printStackTrace();
                throw new FlowRunningException("更新时查询原数据失败:请检查是否有主键或者id列"+ e.getMessage());
            }
        } else {
            return null;
        }
    }

    private DatabaseConfig getDatabaseConfig(DatabaseParamDTO databaseParamDTO) {
        String driver = databaseParamDTO.getDriver();
        String url = databaseParamDTO.getUrl();
        String userName = databaseParamDTO.getUserName();
        String pwd = databaseParamDTO.getPwd();
        String schema = databaseParamDTO.getSchema() == null? "public": databaseParamDTO.getSchema();
        String tableName = databaseParamDTO.getTableName();
        String dbName = databaseParamDTO.getDbName();
        //构造配置
        DatabaseConfig.DatabaseConfigBuilder builder = DatabaseConfig
                .DatabaseConfigBuilder
                .aDatabaseConfig()
                .withType(DataBase.getDataBaseByDriver(driver))
                .withUrl(url)
                .withSchema(schema)
                .withUserName(userName)
                .withDbName(dbName)
                .withUserPwd(pwd)
                .withDriver(driver);

        //设置表信息
        if(StrUtil.isNotEmpty(tableName)){
            builder.withTableName(tableName);
        }
        //重新生成URL
        return  builder.build();
    }
}
